package com.atguigu.flink.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.SqlDialect;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/27 16:34
 */
public class Flink15_SQL_HiveCatalog {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String catalogName = "myhive";  // catalog的名称
        String defaultDatabase = "flinktest";   // 连接上 Hive，默认的数据库
        String hiveConfDir = "F:\\atguigu\\01_course\\code\\hive-conf"; // hive配置文件目录

        // 1、创建一个HiveCatalog
        HiveCatalog hiveCatalog = new HiveCatalog(catalogName, defaultDatabase, hiveConfDir);
        // 2、注册 HiveCatalog
        tableEnv.registerCatalog(catalogName, hiveCatalog);

        // 3、 指定 HiveCatalog
        tableEnv.useCatalog(catalogName);
        // 3、 指定数据库名(如果不指定，就是用前面定义的默认库)
//        tableEnv.useDatabase("hive的库名");

        // 4. 指定 SQL语法 为 Hive的语法
        tableEnv.getConfig().setSqlDialect(SqlDialect.HIVE);

        // 5. 通过 flinksql 操作 hive表
//        tableEnv.executeSql("insert into test values (3,'java',18)");
        tableEnv.executeSql("select * from test").print();

    }
}
